#include <fcntl.h>
#include <string.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <sys/epoll.h>
#include <sys/types.h>
#include <sys/stat.h>

#include <string>

#include "threadpool.h"
#include "epoll_reactor.h"

pthread_mutex_t mutex_cout = PTHREAD_MUTEX_INITIALIZER;

void error_handling(const char *message)
{
    perror(message);
    exit(1);
}

void reactor_run()
{
    // 创建线程池
    threadpool_t *pool = threadpool_create(10, 4, 16);

    int lfd = init_socket(6688);       // 创建监听套接字
    int flag = fcntl(lfd, F_GETFL, 0); // 设置lfd非阻塞
    fcntl(lfd, F_SETFL, flag | O_NONBLOCK);

    // reactor对象,保存了epoll跟节点和rev事件数组
    reactor reac;
    reac.rev_list = new revent[MAX_EPOLL_EVENTS];
    memset(reac.rev_list, 0, (MAX_EPOLL_EVENTS) * sizeof(revent));
    for (int i = 0; i < MAX_EPOLL_EVENTS; i++)
    {
        reac.rev_list[i].recvdata = nullptr;
        reac.rev_list[i].senddata = nullptr;
        reac.rev_list[i].recvdata = new char[1024];
        reac.rev_list[i].senddata = new char[1024];
    }

    reac.epoll_fd = epoll_create(1);
    if (reac.epoll_fd == -1)
        error_handling("create epoll fd error");

    // 为lfd设置写入事件并添加到epoll树
    set_revent(&reac.rev_list[lfd], lfd, accept_connection, &reac);
    // 将fd当下标存放元素
    add_revent(&reac, EPOLLIN | EPOLLET, &reac.rev_list[lfd]);

    epoll_event ev_ready[MAX_EPOLL_EVENTS]; // epoll就绪列表
    while (1)
    {
        // std::cout << "epoll waiting..." << std::endl;
        int ret = epoll_wait(reac.epoll_fd, ev_ready, 1024, -1);
        if (ret == -1 && errno != EINTR) // EINTR表示被信号正常中断
        {
            perror("epoll_wait error");
            exit(1);
        }

        for (int i = 0; i < ret; i++)
        {
            revent *rev = (revent *)ev_ready[i].data.ptr;

            if (rev && rev->callback)
            {
                if (rev->fd == lfd)
                    // 来自监听套接字的新连接同步处理
                    rev->callback(rev->fd, rev->arg);
                else
                    // 其他的任务交由线程池处理
                    threadpool_addtask(rev->callback, rev->fd, rev->arg, pool);
            }
        }
    }

    close(reac.epoll_fd);
    close(lfd);

    // 清理内存
    if (reac.rev_list)
    {
        for (int i = 0; i < MAX_EPOLL_EVENTS; i++)
        {
            if (reac.rev_list[i].recvdata)
                delete[] reac.rev_list[i].recvdata;
            if (reac.rev_list[i].senddata)
                delete[] reac.rev_list[i].senddata;
        }
        delete[] reac.rev_list;
    }

    threadpool_destroy(pool);
}

// 初始化套接字
int init_socket(unsigned short port)
{
    int listen_fd = socket(PF_INET, SOCK_STREAM, 0);
    if (listen_fd == -1)
        error_handling("create socket error");

    sockaddr_in serv_addr;
    memset(&serv_addr, 0, sizeof(serv_addr));
    serv_addr.sin_family = AF_INET;
    serv_addr.sin_addr.s_addr = htonl(INADDR_ANY);
    serv_addr.sin_port = htons(6688);

    // 端口复用
    int option = 1;
    setsockopt(listen_fd, SOL_SOCKET, SO_REUSEADDR, &option, sizeof(option));

    if (bind(listen_fd, (sockaddr *)&serv_addr, sizeof(serv_addr)) == -1)
        error_handling("bind socket error");

    if (listen(listen_fd, 1000) == -1)
        error_handling("listen socket error");

    return listen_fd;
}

void set_revent(revent *rev, int fd, void (*callback)(int, void *), void *arg)
{
    if (!rev)
        return;

    rev->fd = fd;
    rev->callback = callback;
    rev->arg = arg;
}

void add_revent(reactor *reac, uint32_t events, revent *rev)
{
    if (!rev)
        return;

    epoll_event ev;
    ev.events = events;
    ev.data.ptr = rev;
    if (!rev->status)
    {
        epoll_ctl(reac->epoll_fd, EPOLL_CTL_ADD, rev->fd, &ev);
        rev->status = true;
    }
    else
    {
        epoll_ctl(reac->epoll_fd, EPOLL_CTL_MOD, rev->fd, &ev);
    }
}

void remove_revent(reactor *reac, revent *rev)
{
    epoll_ctl(reac->epoll_fd, EPOLL_CTL_DEL, rev->fd, NULL);
    rev->status = 0; // 将该rev再次设为不可用
    reac->clients.erase(rev->fd);
    close(rev->fd);
}

void accept_connection(int fd, void *arg)
{
    // std::cout << "accepting..." << std::endl;
    reactor *reac = (reactor *)arg;  // rev_list首地址
    revent rev = reac->rev_list[fd]; // 当前事件对应的事件信息
    int lfd = rev.fd;
    while (1)
    {
        sockaddr_in addr_clt;
        socklen_t addr_len = sizeof(addr_clt);
        int cfd = accept(lfd, (sockaddr *)&addr_clt, &addr_len);
        if (cfd == -1)
        {
            if (errno == EAGAIN)
                break;
            error_handling("accept socket error");
        }

        // 测试
        // int outfd = open("test2.out", O_WRONLY | O_APPEND);
        // std::string str = "new client cfd=" + std::to_string(cfd) + "\n";
        // write(outfd, str.c_str(), str.size());
        // close(outfd);

        // 将新连接的客户端存入map，默认以ip地址作为用户名
        reac->clients[cfd] = inet_ntoa(addr_clt.sin_addr);

        pthread_mutex_lock(&mutex_cout);
        std::cout << "new client: cfd=" << cfd << ", ip="
                  << inet_ntoa(addr_clt.sin_addr) << "\n"
                  << std::flush;
        pthread_mutex_unlock(&mutex_cout);

        // 设置cfd非阻塞
        int flag = fcntl(cfd, F_GETFL, 0);
        fcntl(cfd, F_SETFL, flag | O_NONBLOCK);

        // 设置并向epoll树添加输入事件
        set_revent(&reac->rev_list[cfd], cfd, read_fd, reac);
        add_revent(reac, EPOLLIN | EPOLLET, &reac->rev_list[cfd]);
    }
}

void read_fd(int fd, void *arg)
{
    reactor *reac = (reactor *)arg;    // rev_list首地址
    revent *rev = &reac->rev_list[fd]; // 当前事件对应的事件信息
    int cfd = rev->fd;

    int ret = 0;
    std::string recv_str;
    char buf[10];
    while (1)
    {
        ret = read(cfd, buf, 10);
        if (ret == -1)
        {
            if (errno == EAGAIN)
            {
                break;
            }
            pthread_mutex_lock(&mutex_cout);
            std::cout << "client[cfd=" << cfd << "] " << std::flush;
            perror("read error");
            pthread_mutex_unlock(&mutex_cout);
            remove_revent(reac, rev);
            return;
        }
        else if (ret == 0)
        {
            pthread_mutex_lock(&mutex_cout);
            std::cout << "client[cfd=" << cfd << "] exit.\n" << std::flush;
            pthread_mutex_unlock(&mutex_cout);
            remove_revent(reac, rev);
            return;
        }
        else
        {
            recv_str.append(buf, ret);
        }
    }

    // 将数据拷贝到接收区
    rev->recv_len = recv_str.size();
    memcpy(rev->recvdata, recv_str.c_str(), rev->recv_len);

    // 业务处理：识别用户并保存用户名
    std::string user_name;
    user_name.append(rev->recvdata, 30);

    if (!strcmp("", user_name.c_str()))
    {
        user_name = reac->clients[cfd];
        memcpy(rev->recvdata, user_name.c_str(), 30);
    }
    else if (reac->clients[cfd] != user_name)
    {
        reac->clients[cfd] = user_name;
    }

    // 打印日志
    // std::cout << "clent[cfd=" << cfd << "]: " << recv_str << std::endl;

    // 除自己外的所有客户端设置写事件并添加到epoll树
    for (auto p : reac->clients)
    {
        if (p.first != cfd)
        {
            revent *tar_rev = &reac->rev_list[p.first];

            // 将消息拷贝到待发区
            tar_rev->send_len = rev->recv_len;
            memcpy(tar_rev->senddata, rev->recvdata, rev->recv_len);

            set_revent(tar_rev, p.first, write_fd, reac);
            add_revent(reac, EPOLLOUT | EPOLLET, tar_rev);
        }
    }

    // 自己继续设置读事件
    // set_revent(rev, cfd, read_fd, reac);
    // add_revent(reac->epoll_fd, EPOLLIN | EPOLLET, rev);
}

void write_fd(int fd, void *arg)
{
    reactor *reac = (reactor *)arg;    // rev_list首地址
    revent *rev = &reac->rev_list[fd]; // 当前事件对应的事件信息
    int cfd = rev->fd;

    // test
    // char test_str[] = "abcdefghijklmn123456789";
    while (1)
    {
        int ret = write(cfd, rev->senddata, rev->send_len);
        if (ret == -1)
        {
            if (errno == EAGAIN)
                continue;
            perror("write error");
            epoll_ctl(reac->epoll_fd, EPOLL_CTL_DEL, cfd, NULL);
            memset(rev, 0, sizeof(revent));
            close(cfd);
            return;
        }
        else
            break;
    }

    pthread_mutex_lock(&mutex_cout);
    std::cout << "send message to client[cfd="
              << cfd << "].\n" << std::flush;
    pthread_mutex_unlock(&mutex_cout);

    set_revent(rev, cfd, read_fd, reac);
    add_revent(reac, EPOLLIN | EPOLLET, rev);
}
